package indi.lqzhi.basic.tools.socket;



import androidx.annotation.WorkerThread;

import com.blankj.utilcode.util.ArrayUtils;
import com.blankj.utilcode.util.LogUtils;
import com.blankj.utilcode.util.StringUtils;
import com.blankj.utilcode.util.ThreadUtils;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.TimerTask;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import indi.lqzhi.basic.tools.socket.message.IMessage;
import indi.lqzhi.basic.tools.socket.message.impl.BytesMessage;
import indi.lqzhi.basic.tools.socket.message.impl.StrMessage;
import indi.lqzhi.basic.tools.thirdparty.ThreadTaskTools;

import static indi.lqzhi.basic.tools.socket.ISocketClient.OnConnectLinstener;
import static indi.lqzhi.basic.tools.socket.ISocketClient.OnDisConnectLinstener;
import static indi.lqzhi.basic.tools.socket.ISocketClient.OnResponseDataLinstener;
import static indi.lqzhi.basic.tools.socket.ISocketClient.OnResponseLinstener;

public final class SocketClient {


    /************************** 各个变量和标志位 **********************************/

    /**
     * SocketChannel
     */
    private volatile SocketChannel socketChannel;

    /**
     * 分发器
     */
    private volatile Selector selector;

    /**
     * Socket服务器地址
     */
    private volatile  SocketAddress address;

    /**
     * Socket的连接状态 true代表能够发送数据(已连接)， false代表不能发送数据(未连接)
     */
    private volatile boolean socketConnectStatus;

    /**
     * 发送的字符集
     */
    private volatile Charset sendCharset;

    /**
     * 接收的字符集
     */
    private volatile Charset recvCharset;

    /**
     * 分发器线程停止位
     */
    public volatile boolean selectorThreadStopFlag;

    /**
     * 接收的子线程停止位
     */
    public volatile boolean recvThreadStopFlag;

    /**
     * 发送的队列
     */
    private volatile LinkedBlockingQueue<IMessage> sendQueue;

    /**
     * 接收的队列
     */
    private volatile LinkedBlockingDeque<IMessage> recvQueue;

    private volatile TimerTask timerTask = new TimerTask() {
        @Override
        public void run() {

        }
    };

    /**
     * 接收数据的buffer
     */
    private ByteBuffer buffer = ByteBuffer.allocate(2048);


    /************************** 各个监听器 **********************************/

    /**
     * 连接监听器
     */
    private volatile List<OnConnectLinstener> connectLinstenerList;

    /**
     * 断开连接监听器
     */
    private volatile List<OnDisConnectLinstener> disConnectLinstenerList;

    /**
     * 回调处理监听器
     */
    private volatile List<OnResponseLinstener> responseLinstenerList;

    /**
     * 回调编码后的数据处理监听器
     */
    private volatile List<OnResponseDataLinstener> responseDataLinstenerList;


    /************************** 各个运行器 **********************************/

    private final Runnable selectorRunnable = () -> {
        while (!selectorThreadStopFlag){
            try{
                //写入检查
                IMessage message = sendQueue.poll();
                if (message != null && socketChannel != null){
                    //发送
                    ByteBuffer byteBuffer = message.getByteBuffer();
                    if (byteBuffer != null){
                        socketChannel.write(byteBuffer);
                    }
                }
                //轮询
                if (selector.selectNow() == 0) {
                    //为0，继续轮询
                    continue;
                }
                //轮询事件
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()){
                    SelectionKey key = iterator.next();
                    iterator.remove();//移除当前key，避免重复操作

                    //处理key是可读的的事件
                    if(key.isReadable()) {
                        SocketChannel socketChannel = (SocketChannel) key.channel();
                        buffer.clear();//清楚Buffer
                        int length = socketChannel.read(buffer);
                        if (length == -1){
                            LogUtils.d("服务器已断开~");
                            key.cancel();//取消事件
                            throw new SocketException("读取错误，可能服务器已断开。");//读错误，抛出异常 关闭socket
                        }

                        if (length == 0) {
                            LogUtils.d("接收到服务器的数据为空~");
                            continue;
                        }
                        //放入队列
                        byte[] bytes = new byte[length];
                        System.arraycopy(buffer.array(),0,bytes,0,length);
                        recvQueue.offer(new BytesMessage(bytes));
                        continue;//处理完了就继续迭代
                    }
                }
            }catch (Exception exception){
                closeSocket();//强制关闭Socket
                handlerOnDisConnectLinstener(exception);//处理断开的事件
            }
        }
    };

    //接收的子线程
    private final Runnable recvRunnable = () -> {
        while (!recvThreadStopFlag){
            try {
                BytesMessage message = (BytesMessage) recvQueue.poll();
                if (message == null){
                    continue;
                }
                byte[] bytes = message.getBytes();
                //处理子线程的数据
                handlerOnResponseLinstener(bytes);
                //转换成response
                String response = new String(bytes,0,bytes.length, StandardCharsets.US_ASCII);
                LogUtils.d("接收到了服务器的消息："+response);
                if (StringUtils.isTrimEmpty(response)){
                    continue;//继续
                }
                //处理子线程的数据
                handlerOnResponseDataLinstener(response);
                //处理数据

            }catch (Exception exception){
                closeSocket();
                handlerOnDisConnectLinstener(exception);
            }
        }
    };

    /**
     * 连接的子线程
     */
    private final Runnable connectRunnable = () -> {
        try {
            this.selector = Selector.open();
            socketChannel = SocketChannel.open();//打开连接
            socketChannel.configureBlocking(false);//非堵塞方式
            socketChannel.connect(address);
            int timeout = 6;
            while(!socketChannel.finishConnect()){
                //wait, or do something else...
                Thread.sleep(1000);
                timeout--;
                if (timeout < 0){
                    closeSocket();
                    throw new SocketTimeoutException("已经超时6秒了...连接失败！");
                }
            }
            socketChannel.register(selector, SelectionKey.OP_READ);
            if (socketChannel.isConnected()) {
                handlerOnConnectLinstenerSuccess();
                socketConnectStatus = true;
            } else {
                handlerOnConnectLinstenerFail(null);
                socketConnectStatus = false;
                return;//不开启线程
            }
        } catch (IOException | InterruptedException exception) {
            handlerOnConnectLinstenerFail(exception);
            socketConnectStatus = false;
            return;//不开启线程
        }
        //开启socket接收和发送线程
        ThreadUtils.executeByIoWithDelay(ThreadTaskTools.getTask(selectorRunnable), 1, TimeUnit.SECONDS);
        ThreadUtils.executeByIoWithDelay(ThreadTaskTools.getTask(recvRunnable),1,TimeUnit.SECONDS);
        //ThreadUtils.executeByIoWithDelay(ThreadTaskUtils.getTask(socketRecvRunable), 1, TimeUnit.SECONDS);
    };


    /************************** 构造对象区 **********************************/

    /**
     * 私有的方法限制对象产生
     */
    private SocketClient(SocketAddress address){
        this.address = Objects.requireNonNull(address);

        this.sendCharset = StandardCharsets.UTF_8;//默认为UTF-8
        this.recvCharset = StandardCharsets.UTF_8;//默认为UTF-8
        this.sendQueue = new LinkedBlockingQueue<>();//创建发送的消息队列
        this.recvQueue = new LinkedBlockingDeque<>();//创建接收的消息队列

        //监听器的初始化
        this.connectLinstenerList = new CopyOnWriteArrayList<>();
        this.disConnectLinstenerList = new CopyOnWriteArrayList<>();
        this.responseLinstenerList = new CopyOnWriteArrayList<>();
        this.responseDataLinstenerList = new CopyOnWriteArrayList<>();

        //连接状态：未连接
        socketConnectStatus = false;
        selectorThreadStopFlag = false;
        recvThreadStopFlag = false;
    }

    /**
     * 提供静态方法创建对象
     * @return SocketClient对象
     */
    public static SocketClient create(InetAddress address,int port){
        return new SocketClient(new InetSocketAddress(address,port));
    }

    /**
     * 提供静态方法创建对象
     * @return SocketClient对象
     */
    public static SocketClient create(String host,int port){
        return new SocketClient(new InetSocketAddress(host,port));
    }

    /************************** 方法区 **********************************/

    /**
     * 改变服务器的host和端口
     * @param address 地址
     * @param port 端口
     * @return true 改变成功 false 改变失败
     */
    public synchronized boolean changeServerHostAndPort(InetAddress address,int port){
        //当是在连接状态下，不能改变
        if (socketConnectStatus){
            return false;
        }
        if (address == null){
            return false;
        }
        //检验非法端口
        if (port < 0 || port > 65535){
            throw new IllegalArgumentException("Illegal Port");
        }

        this.address = new InetSocketAddress(address,port);

        //关闭socket
        closeSocket();

        socketConnectStatus = false;
        selectorThreadStopFlag = false;//关闭
        recvThreadStopFlag = false;
        return true;
    }

    /**
     * 改变服务器的host和端口
     * @param host 地址
     * @param port 端口
     * @return true 改变成功 false 改变失败
     */
    public synchronized boolean changeServerHostAndPort(String host,int port) {
        //当是在连接状态下，不能改变
        if (socketConnectStatus){
            return false;
        }

        if (StringUtils.isTrimEmpty(host)){
            return false;
        }

        //检验非法端口
        if (port < 0 || port > 65535){
            throw new IllegalArgumentException("Illegal Port");
        }

        this.address = new InetSocketAddress(host,port);

        //关闭Socket
        closeSocket();


        socketConnectStatus = false;
        selectorThreadStopFlag = false;//关闭
        recvThreadStopFlag = false;
        return true;
    }

    /**
     * 向服务器发起连接
     */
    public synchronized void connect(){
        //如果未连接，则运行发起请求，否则抛出异常
        if (!socketConnectStatus){
            recvThreadStopFlag = false;
            selectorThreadStopFlag = false;
            ThreadUtils.executeByIo(ThreadTaskTools.getTask(connectRunnable));
            return;
        }
        throw new UnsupportedOperationException("socket is connect!");
    }

    /**
     * 当发送消息队列满时抛出异常
     * @param message 消息
     * @return true(已加入到发送消息队列中，等待发送) false(添加到消息队列错误 或 字符串为空)
     */
    public synchronized boolean sendMessage(String message) throws IllegalStateException{
        if (!socketConnectStatus || StringUtils.isEmpty(message)){
            return false;
        }
        return sendQueue.add(new StrMessage(message));
    }

    /**
     * 当发送消息队列满时会堵塞，所以必须要在工作线程中运行！
     * @param message 消息
     * @return true(已加入到发送消息队列中，等待发送) false(添加到消息队列错误 或 字符串为空)
     */
    @WorkerThread
    public synchronized boolean putSendMessage(String message) throws InterruptedException {
        if (!socketConnectStatus || StringUtils.isEmpty(message)){
            return false;
        }
        sendQueue.put(new StrMessage(message));
        return true;
    }

    /**
     * 当发送消息队列满时返回false
     * @param message 消息
     * @return true(已加入到发送消息队列中，等待发送) false(添加到消息队列错误 或 字符串为空)
     */
    public synchronized boolean offerSendMessage(String message){
        if (!socketConnectStatus || StringUtils.isEmpty(message)){
            return false;
        }
        return sendQueue.offer(new StrMessage(message));
    }


    /**
     * 当发送消息队列满时抛出异常
     * @param message 消息
     * @return true(已加入到发送消息队列中，等待发送) false(添加到消息队列错误 或 字符串为空)
     */
    public synchronized boolean sendMessage(byte[] message) throws IllegalStateException{
        if (!socketConnectStatus || ArrayUtils.isEmpty(message)){
            return false;
        }
        return sendQueue.add(new BytesMessage(message));
    }

    /**
     * 当发送消息队列满时会堵塞，所以必须要在工作线程中运行！
     * @param message 消息
     * @return true(已加入到发送消息队列中，等待发送) false(添加到消息队列错误 或 字符串为空)
     */
    @WorkerThread
    public synchronized boolean putSendMessage(byte[] message) throws InterruptedException {
        if (!socketConnectStatus || ArrayUtils.isEmpty(message)){
            return false;
        }
        sendQueue.put(new BytesMessage(message));
        return true;
    }

    /**
     * 当发送消息队列满时返回false
     * @param message 消息
     * @return true(已加入到发送消息队列中，等待发送) false(添加到消息队列错误 或 字符串为空)
     */
    public synchronized boolean offerSendMessage(byte[] message){
        if (!socketConnectStatus || ArrayUtils.isEmpty(message)){
            return false;
        }
        return sendQueue.offer(new BytesMessage(message));
    }

    /**
     * 强制关闭Socket，此方法会如果关不掉，就不管他了
     */
    public synchronized void closeSocket(){
        recvThreadStopFlag = true;
        selectorThreadStopFlag = true;//关闭
        try{
            if (socketChannel != null) {
                socketChannel.close();//断开连接
            }
            socketChannel = null;
            if (selector != null){
                selector.close();
            }
            selector = null;
        }catch (Exception exception){
            LogUtils.e("关闭SocketChannel异常：",exception);
            socketChannel = null;
            selector = null;
        }
        socketConnectStatus = false;
    }
    /**
     * 断开连接，此方法和closeSocket的区别是，他会回调disconnect的事件，而closeSocket不会
     */
    public synchronized void disconnect(){
        closeSocket();
        handlerOnDisConnectLinstener(null);
    }


    public boolean isSocketConnect() {
        return socketConnectStatus;
    }

    public Charset getSendCharset() {
        return sendCharset;
    }

    public Charset getRecvCharset() {
        return recvCharset;
    }

    public void setSendCharset(Charset sendCharset) {
        this.sendCharset = sendCharset;
    }

    public void setRecvCharset(Charset recvCharset) {
        this.recvCharset = recvCharset;
    }

    public SocketAddress getAddress() {
        return address;
    }

    public void setAddress(SocketAddress address) {
        this.address = address;
    }


    /************************** 监听器注册销毁方法区 **********************************/

    /**
     * 注册连接监听器
     * @param onConnectLinstener 需要注册的连接的监听器
     * @return true 注册成功 false 注册失败
     */
    public boolean registerOnConnectLinstener(OnConnectLinstener onConnectLinstener){
        if (onConnectLinstener == null){
            return false;
        }
        return connectLinstenerList.add(onConnectLinstener);
    }

    /**
     * 销毁指定的连接监听器
     * @param onConnectLinstener 需要销毁的连接的监听器
     * @return true 销毁成功 false 销毁失败
     */
    public boolean unregisterOnConnectLinstener(OnConnectLinstener onConnectLinstener){
        if (onConnectLinstener == null){
            return false;
        }
        return connectLinstenerList.remove(onConnectLinstener);
    }

    /**
     * 注册断开的监听器
     * @param onDisConnectLinstener 需要注册的断开的监听器
     * @return true 注册成功 false 注册失败
     */
    public boolean registerOnDisConnectLinstener(OnDisConnectLinstener onDisConnectLinstener){
        if (onDisConnectLinstener == null){
            return false;
        }
        return disConnectLinstenerList.add(onDisConnectLinstener);
    }

    /**
     * 销毁指定的断开的监听器
     * @param onDisConnectLinstener 需要销毁的断开的监听器
     * @return true 销毁成功 false 销毁失败
     */
    public boolean unregisterOnDisConnectLinstener(OnDisConnectLinstener onDisConnectLinstener){
        if (onDisConnectLinstener == null){
            return false;
        }
        return disConnectLinstenerList.remove(onDisConnectLinstener);
    }

    /**
     * 注册原始回调的监听器
     * @param onResponseLinstener 需要注册的 处理元素数据的监听器
     * @return true 注册成功 false 注册失败
     */
    public boolean registerOnResponseLinstener(OnResponseLinstener onResponseLinstener){
        if (onResponseLinstener == null){
            return false;
        }
        return responseLinstenerList.add(onResponseLinstener);
    }

    /**
     * 销毁指定原始回调的监听器
     * @param onResponseLinstener 需要销毁的 处理元素数据的监听器
     * @return true 销毁成功 false 销毁失败
     */
    public boolean unregisterOnResponseLinstener(OnResponseLinstener onResponseLinstener){
        if (onResponseLinstener == null){
            return false;
        }
        return responseLinstenerList.remove(onResponseLinstener);
    }

    /**
     * 注册回调的监听器
     * @param onResponseDataLinstener 需要注册的 处理元素数据的监听器
     * @return true 注册成功 false 注册失败
     */
    public boolean registerOnResponseDataLinstener(OnResponseDataLinstener onResponseDataLinstener){
        if (onResponseDataLinstener == null){
            return false;
        }
        return responseDataLinstenerList.add(onResponseDataLinstener);
    }

    /**
     * 销毁指定回调的监听器
     * @param onResponseDataLinstener 需要销毁的 处理元素数据的监听器
     * @return true 销毁成功 false 销毁失败
     */
    public boolean unregisterOnResponseDataLinstener(OnResponseDataLinstener onResponseDataLinstener){
        if (onResponseDataLinstener == null){
            return false;
        }
        return responseDataLinstenerList.remove(onResponseDataLinstener);
    }

    /************************** 处理监听器方法区 **********************************/

    /**
     * 处理连接成功的监听器
     */
    private void handlerOnConnectLinstenerSuccess(){
        for (OnConnectLinstener onConnectLinstener : connectLinstenerList) {
            if (onConnectLinstener == null){
                continue;
            }
            if (onConnectLinstener.isMainThread()){
                //回调主线程去处理
                ThreadUtils.runOnUiThread(() -> onConnectLinstener.onConnectSuccess(SocketClient.this));
            }else {
                ThreadUtils.executeByCached(ThreadTaskTools.getTask(() -> onConnectLinstener.onConnectSuccess(SocketClient.this)));
            }

        }
    }

    /**
     * 处理连接失败的监听器
     */
    private void handlerOnConnectLinstenerFail(Throwable throwable){
        for (OnConnectLinstener onConnectLinstener : connectLinstenerList) {
            if (onConnectLinstener == null){
                continue;
            }
            if (onConnectLinstener.isMainThread()){
                //回调主线程去处理
                ThreadUtils.runOnUiThread(() -> onConnectLinstener.onConnectFail(SocketClient.this,throwable));
            }else {
                //取消回调，使用子线程去处理
                ThreadUtils.executeByCached(ThreadTaskTools.getTask(() -> onConnectLinstener.onConnectFail(SocketClient.this,throwable)));
            }
        }
    }

    /**
     * 处理连接失败的监听器
     */
    private void handlerOnDisConnectLinstener(Throwable throwable){
        for (OnDisConnectLinstener onDisConnectLinstener : disConnectLinstenerList) {
            if (onDisConnectLinstener == null){
                continue;
            }
            if (onDisConnectLinstener.isMainThread()){
                //回调主线程去处理
                ThreadUtils.runOnUiThread(() -> onDisConnectLinstener.onDisConnect(throwable));
            }else {
                //取消回调，使用子线程去处理
                ThreadUtils.executeByCached(ThreadTaskTools.getTask(() -> onDisConnectLinstener.onDisConnect(throwable)));
            }
        }
    }

    /**
     * 处理原始回调接收到的数据的监听器
     */
    private void handlerOnResponseLinstener(final byte[] response){
        for (OnResponseLinstener onResponseLinstener : responseLinstenerList) {
            if (onResponseLinstener == null){
                continue;
            }
            if (onResponseLinstener.isMainThread()){
                //回调主线程去处理
                ThreadUtils.runOnUiThread(() -> onResponseLinstener.onResponse(SocketClient.this,response));
            }else {
                //取消回调，使用子线程去处理
                ThreadUtils.executeByCached(ThreadTaskTools.getTask(() -> onResponseLinstener.onResponse(SocketClient.this,response)));
            }
        }
    }

    /**
     * 处理接收到的数据的监听器
     */
    private void handlerOnResponseDataLinstener(final String response){
        for (OnResponseDataLinstener onResponseDataLinstener : responseDataLinstenerList) {
            if (onResponseDataLinstener == null){
                continue;
            }
            if (onResponseDataLinstener.isMainThread()){
                //回调主线程去处理
                ThreadUtils.runOnUiThread(() -> onResponseDataLinstener.onResponse(SocketClient.this,response));
            }else {
                //取消回调，使用子线程去处理
                ThreadUtils.executeByCached(ThreadTaskTools.getTask(() -> onResponseDataLinstener.onResponse(SocketClient.this,response)));
            }

        }
    }



}
